package com.bigdata.warehouse.sink;

import com.bigdata.warehouse.entity.CarBehavior;
import com.bigdata.warehouse.entity.OwnerBehavior;
import com.bigdata.warehouse.util.ClickhouseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;

public class DwsOwnerBehaviorClickhouseSink extends RichSinkFunction<OwnerBehavior> {
    Connection connection =null;
    String sql;
    public DwsOwnerBehaviorClickhouseSink(String sql){
        this.sql=sql;
    }
    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("11111111111111111111111111111111111111111");
        connection= ClickhouseUtil.getConn("slave3",8123,"default");
    }

    @Override
    public void close() throws Exception {
        if(connection !=null){
            connection.close();
        }
    }

    @Override
    public void invoke(OwnerBehavior ob, Context context) throws Exception {
        System.out.println(ob.getOwnerId()+"--------------------------------------");
        System.out.println("22222222222222222222222222222222222222222222");
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        preparedStatement.setInt(1,ob.getOwnerId());
        preparedStatement.setString(2,ob.getOwerSex());
        preparedStatement.setInt(3,ob.getOwerAge());
        preparedStatement.setString(4,ob.getOwerHomeplace());
        preparedStatement.setString(5,ob.getOwerWorkplace());
        preparedStatement.setInt(6,ob.getCommunity_id());
        preparedStatement.setString(7,ob.getCommunityName());
        preparedStatement.setInt(8,ob.getCity_id());
        preparedStatement.setInt(9,ob.getDistrict_id());
        preparedStatement.setString(10,ob.getDistrict_name());
        preparedStatement.setInt(11,ob.getEntrance_card_cnt());
        preparedStatement.setInt(12,ob.getEntrance_pw_cnt());
        preparedStatement.setInt(13,ob.getEntrance_face_cnt());
        preparedStatement.setInt(14,ob.getEntrance_remote_cnt());
        preparedStatement.setTimestamp(15,ob.getStt());
        preparedStatement.setTimestamp(16,ob.getEdt());
        preparedStatement.setTimestamp(17,ob.getTs());
        preparedStatement.addBatch();
        preparedStatement.executeBatch();
        connection.commit();
    }
}
